package com.zhuiwei.bigdata.hdfs.client;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.net.URI;

/**
 * HDFS客户端操作
 * @author lin
 * @version 2019/04/07
 */
public class HdfsClient {

    private static  final String HDFS_PATH="hdfs://192.168.154.101:9000";
    private static final String HADOOP_USER_NAME="linmh";

    /**
     * 上传文件
     * @param src 指要上传的文件路径
     * @param dst 指将文件上传到的路径
     * @throws Exception 异常
     */
    public static void copyFromLocalFile(String src, String dst)throws Exception {
        //获取文件系统
        Configuration configuration=new Configuration();
        // configuration.set("fs.defaultFS", "hdfs://hadoop102:9000");

        //客户端去操作hdfs时，是有一个用户身份的,默认情况下，hdfs客户端api会从jvm中获取一个参数来作为自己的用户身份：-DHADOOP_USER_NAME=hadoop
        //FileSystem hdfs = FileSystem.get(getHdfsConfig());
        //也可以在构造客户端fs对象时，通过参数传递进去。FileSystem.get(路径，配置对象，用户名);
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //上传文件
        fs.copyFromLocalFile(new Path(src),new Path(dst));

        //关闭资源
        fs.close();
    }

    /**
     * 下载文件
     * @param delSrc 指是否将原文件删除
     * @param src 指要下载的文件路径
     * @param dst 指将文件下载到的路径
     * @param useRawLocalFileSystem 是否开启文件校验
     * @throws Exception 异常
     */
    public static void copyToLocalFile(boolean delSrc, String src, String dst, boolean useRawLocalFileSystem) throws Exception {
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH), configuration,HADOOP_USER_NAME);

        //下载文件
        fs.copyToLocalFile(delSrc,new Path(src),new Path(dst),useRawLocalFileSystem);

        //关闭资源
        fs.close();
    }

    /**
     * 创建目录
     * @param f 指目录路径
     * @throws Exception 异常
     */
    public static void mkdirs(String f)throws Exception{
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH), configuration,HADOOP_USER_NAME);

        //创建目录
        boolean isSuccess=fs.mkdirs(new Path(f));

        //关闭资源
        fs.close();
    }

    /**
     * 删除目录或文件
     * @param var1 指删除的目录或文件
     * @param var2 指是否删除非空的目录
     * @throws Exception 异常
     */
    public static void delete(String var1, boolean var2) throws Exception{
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //删除文件夹
        boolean isSuccess=fs.delete(new Path(var1),var2);

        //关闭资源
        fs.close();
    }

    /**
     * 修改HDFS文件名
     * @param var1 指修改前的文件名
     * @param var2 指修改后的文件名
     * @throws Exception 异常
     */
    public static void rename(String var1, String var2)throws Exception{
        //获取文件
        Configuration configuration=new Configuration();
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //修改文件名称
        boolean isSuccess=fs.rename(new Path(var1),new Path(var2));

        //关闭资源
        fs.close();
    }

    /**
     * 查看HDFS文件详情
     * @param f 指查看的目录
     * @param recursive 指是否递归查询
     * @throws Exception
     */
    public static  void  listFiles( String f, boolean recursive)throws Exception{
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs =FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //获取文件详情
        RemoteIterator<LocatedFileStatus> listFiles= fs.listFiles(new Path(f),recursive);

        while (listFiles.hasNext()){
            LocatedFileStatus status=listFiles.next();
            //文件名称
            System.out.println(status.getPath().getName());
            //长度
            System.out.println(status.getLen());
            //权限
            System.out.println(status.getPermission());
            //组
            System.out.println(status.getGroup());

            //获取存储的块信息
            BlockLocation[] blockLocations=status.getBlockLocations();
            for (BlockLocation blockLocation:blockLocations){

                //获取块存储的主机节点
                String[] hosts=blockLocation.getHosts();

                for (String host:hosts){
                    System.out.println(host);
                }
            }
            System.out.println("--------");

        }

    }

    /**
     * 判断是文件还是文件夹
     * @param var1 指当前文件或文件夹路径
     * @throws Exception 异常
     */
    public static void listStatus(String var1)throws Exception{
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //判断是文件还是文件夹
        FileStatus[] listStatus=fs.listStatus(new Path(var1));

        for (FileStatus fileStatus:listStatus){

            if (fileStatus.isFile()){
                //f --指文件
                System.out.println("f:"+fileStatus.getPath().getName());
            }else if (fileStatus.isDirectory()){
                //d --指目录
                System.out.println("d:"+fileStatus.getPath().getName());
            }
        }

        //关闭资源
        fs.close();
    }

    /**
     * 通过IO流上传HDFS文件
     * @param src 指要上传的文件路径
     * @param dst 指将文件上传到的路径
     * @throws Exception 异常
     */
    public static void putFileToHDFS(String src,String dst)throws Exception{
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs =FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //创建输入流
        FileInputStream fis=new FileInputStream(new File(src));

        //创建输出流
        FSDataOutputStream fos=fs.create(new Path(dst));

        //流拷贝
        IOUtils.copyBytes(fis,fos,configuration);

        //关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    }

    /**
     * 通过IO流下载HDFS文件
     * @param src 指下载的文件路径
     * @param dst 指将文件下载到的路径
     * @throws Exception 异常
     */
    public static void fileFromHDFS(String src, String dst)throws Exception{
        //获取文件系统
        Configuration configuration=new Configuration();
        FileSystem fs=FileSystem.get(new URI(HDFS_PATH),configuration,HADOOP_USER_NAME);

        //创建输入流
        FSDataInputStream fis=fs.open(new Path(src));

        //创建输出流
        FileOutputStream fos=new FileOutputStream(new File(dst));

        //流拷贝
        IOUtils.copyBytes(fis,fos,configuration);

        //关闭资源
        IOUtils.closeStream(fis);
        IOUtils.closeStream(fos);
    }


    public static void main(String[] args) throws Exception {
//        copyFromLocalFile("e:/hello.txt","/目录/hello4.txt");
//        copyToLocalFile(false,"/hello1.txt","e:/helloExport.txt",true);
//        delete("/目录",true);
//        delete("/hello2.txt",true);
//        rename("/hello3.txt","/hello3Rename.txt");
//        listFiles("/",true);
//        listStatus("/hello4.txt");
//        putFileToHDFS("e:/hello.txt","/stream/hello.txt");
//        fileFromHDFS("/hello1.txt","e:/haha.txt");
    }
}
